package org.zn.etl.job.handler;

import static org.zn.etl.job.context.GlobalContextKey.*;

import java.util.Date;
import java.util.List;
import java.util.Map;
import java.util.Objects;
import java.util.Optional;
import java.util.stream.Collectors;

import javax.sql.DataSource;

import org.apache.commons.collections4.CollectionUtils;
import org.apache.commons.lang3.StringUtils;
import org.apache.commons.lang3.tuple.Pair;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.context.ApplicationEventPublisher;
import org.springframework.expression.EvaluationContext;
import org.springframework.stereotype.Component;
import org.springframework.util.StopWatch;
import org.zn.datax.common.entity.ColumnInfo;
import org.zn.datax.common.entity.TableInfo;
import org.zn.etl.datasource.DataSourceFactoryHolder;
import org.zn.etl.jdbc.QueryHelper;
import org.zn.etl.jdbc.SqlBuilder;
import org.zn.etl.job.context.ClusterJobEvent;
import org.zn.etl.job.context.JobExecInstanceContext;
import org.zn.etl.job.context.JobRunningCmd;
import org.zn.etl.job.entity.CheckPoint;
import org.zn.etl.job.entity.DataMapping;
import org.zn.etl.job.entity.EtlJob;
import org.zn.etl.misc.DataSourceInfo2ConfigMap;
import org.zn.etl.misc.ExpressionHelper;
import org.zn.etl.service.CheckPointService;

import lombok.extern.slf4j.Slf4j;

@Component("SqlJobHandler")
@Slf4j
public class SqlJobHandler implements JobHandler {
	

	@Autowired
	private DataSourceFactoryHolder factoryHolder; 
	
	private CheckPointService checkPointService;
	
	@Autowired
	private  ApplicationEventPublisher eventPublisher;

	@Autowired
	private ExpressionHelper expressionHelper; 
	
	@Override
	public int executeJob(JobExecInstanceContext jei) throws Exception {
		
		EtlJob  ej=jei.getJob();
		Map<String, Object> context=jei.getContextMap();
		
		Map<String, String> sourceConfig=DataSourceInfo2ConfigMap.toMap(ej.getSource());		
		Map<String, String> targetConfig=DataSourceInfo2ConfigMap.toMap(ej.getTarget());
		
		DataSource sSource=factoryHolder.getFactory(ej.getSource().getFactoryType()).getDataSource(ej.getSource().getKeyName(), sourceConfig);
		DataSource tSource=factoryHolder.getFactory(ej.getTarget().getFactoryType()).getDataSource(ej.getTarget().getKeyName(), targetConfig);
		
		Long total=0L;
		
		//context.put(EXEC_INS, jei);
		
		if(!Boolean.FALSE.equals(context.get(CHECKPOINTENABLE))){
			loadCheckPoint(jei,context);
		}	 
		
		String writeMode= (String) context.get(WRITEMODE);
		
		String loadMode= (String) context.get(LOADMODE);
	 
		for (DataMapping dataMapping : ej.getDataMappings()) {
			
			String sourceSql= dataMapping.getSourceSql();
			TableInfo srcTable= dataMapping.getSourceTable();
			TableInfo tagTable= dataMapping.getTargetTable();
			List<Pair<ColumnInfo, ColumnInfo>> columnMappingPairs  =  dataMapping.getColumnMappingPairs();
			
			String srcpk=srcTable.getPk();
			context.put(SRC_PK, srcpk.toLowerCase());
			
			String tagTableName= tagTable.getSchema()+"."+tagTable.getTable();

			//List<String> tagColNames=columnMappingPairs.stream().map( it-> it.getRight().getName() ).collect(Collectors.toList());
			
			List<String> srcColNames=columnMappingPairs.stream().map(it-> it.getLeft().getName()).collect(Collectors.toList());
			
			String sourceWhere= (String) context.get("sourceWhere");
			if(StringUtils.isNotEmpty(sourceWhere)) {
				srcTable.setWhere(sourceWhere);
			}
			
			String pageMode= (String) context.get(PAGED_MODE);
			
			if(StringUtils.isEmpty(sourceSql)) {
				sourceSql=SqlBuilder.tableSelectSql(pageMode,loadMode,ej.getSource().getDbType(), srcTable, srcColNames);
			}
			
			Long maxOf= context.get(MAX_OFFSET)!=null? Long.valueOf(context.get(MAX_OFFSET).toString()):null ;
			Long maxRs= context.get(MAXROWS)!=null? Long.valueOf(context.get(MAXROWS).toString()):null;
			
			Long soff= context.get(START_OFFSET)!=null? Long.valueOf(context.get(START_OFFSET).toString()):null;
			
			if(soff==null) {
				soff=0L;
				context.put(START_OFFSET, soff);
			}
			
			Integer psize=(Integer) context.get(PAGED_SIZE);
			if(psize==null) {
				psize=DEF_PAGE_SIZE;
				context.put(PAGED_SIZE,psize );
			}
			
			boolean  notPage=Boolean.FALSE.equals(context.get(PAGED));
			
			WriteAgent wa=new WriteAgent(writeMode, tagTable.getSchema(), tagTable.getTable(), tagTable.getPk(), columnMappingPairs, tSource,ej.getTarget().getDbType());
			
				while (true) {

					if(! checkLoad(jei,total)) return -1;
						
					soff = soff + psize;
					context.put(END_OFFSET,soff);
					
					try {
						int l =0;
						
						if(context.get(BUFFLOAD)!=null) {
							l=rwBufferLoad(sourceSql, jei, context, sSource, tSource, tagTableName, columnMappingPairs, wa);							
						}
						else {
							load(sourceSql, jei,context, sSource, tSource, tagTableName, columnMappingPairs, wa);							
						}
						
						if (l < 0 || notPage) {
							break;
						}

						total += l;
						
						//complate check						
					} 
					catch (WriteErrorOverException e) {

						saveCheckPoint(ej, context, jei);
						context.put(CHECKMODE, CheckMode.ERROR.getMode());
						throw e;
					}

					context.put(TOTALROWS, total);
					context.put(START_OFFSET, soff);
					
					if (!Boolean.FALSE.equals(context.get(CHECKPOINTENABLE))) {
						saveCheckPoint(ej, context, jei);
					}
					
					//custom  bound check
					if(maxOf!=null&& soff >=maxOf) {
						break;
					}
					
					if(maxRs!=null&& total >=maxRs) {
						break;
					}

				}

			logInfo(jei,"load data finish , begin postSQL");
			
			String postSql= dataMapping.getTargetSql();
			
			if(StringUtils.isNotEmpty(postSql)) {
				
				String[] psqls=postSql.split(";");
				
				for (String sql : psqls) {
					
					wrapSplExp(sql, context);
					logInfo(jei,"execute  postSQL:{}",sql);
					
					wa.executeUpdate(sql,null);
					
				}
		
			}

		}
		
		jei.setTotal(total);
		return 1;

	}
	
	
	
	private boolean checkLoad(JobExecInstanceContext jei,Long total){
		
		jei.setTotal(total);
		
		//NodeRunningContextManager Listener
		eventPublisher.publishEvent(new ClusterJobEvent(jei));
	
		if(JobRunningCmd.cancel.equals( jei.getDoCmd())) {			
			logInfo(jei,"job is cancel");
			jei.setStatus(ExecStatus.cannel.getStatus());
			
			return  false;
		}

		
		return  true;	
		
	}
	
	
	private void loadCheckPoint(JobExecInstanceContext jei,Map<String, Object> context) {
		
		jei.setCheckPoints(checkPointService.loadCheckPoint(jei.getId()));
		Integer checkPointNum=(Integer) context.get(CHECKNUM);
		jei.initCheckPoints(checkPointNum);
		
	}
	
	
	
	private void saveCheckPoint(EtlJob  ej,Map<String, Object> context,JobExecInstanceContext jei) {
		
		String checkMode=(String) context.get(CHECKMODE);
		Integer freq=(Integer) context.get(CHECKFREQ);
	
		CheckPoint cp=null;
		
		if(CheckMode.TRUN.getMode().equals(checkMode)) {

			if(freq==null) freq=1;
			
			Integer fc=jei.getCheckPointFreqConut();

			if(fc>=freq) {
				cp=jei.nextCheckPoints();
				jei.setCheckPointFreqConut(1);
				Map<String, Object>  crd =(Map<String, Object>) context.get(CHECKROWDATA);
				if(Objects.nonNull(crd) ) {					
					cp.setCheckedRowid(  Optional.ofNullable(crd.get(context.get(SRC_PK))).orElse("").toString());
				}
			}else {
				jei.setCheckPointFreqConut(++fc);
			}
		}
		else if(CheckMode.ERROR.getMode().equals(checkMode)) {
			cp=new CheckPoint();
			cp.setCheckedRowid( context.get(ERROR_DATA_ID).toString());
		}
		
		if(cp!=null) {
			
		Date ct=new Date();
		cp.setCheckTime(ct);
		cp.setUpdateTime(ct);
		if(cp.getCreateTime()==null) {
			cp.setCreateTime(ct);
		}
		
		cp.setMode(checkMode);

		cp.setStartOffset((Long) context.get(START_OFFSET));
		cp.setEndOffset((Long) context.get(END_OFFSET));
		cp.setWritedRows((Long)context.get(TOTALROWS));
		
		checkPointService.saveCheckPoint(cp);
		}
		
		
	}
	
	

	private Integer doWrite(JobExecInstanceContext ins,List<Map<String, Object>>  list,String pageMode,Map<String, Object> context,WriteAgent writeAgent,List<Pair<ColumnInfo, ColumnInfo>> columnMappingPairs) throws WriteErrorOverException {
		
		logDebug(ins,"select rows:{}",list.size());
		
		if(CollectionUtils.isEmpty(list)) {
			return -1;
		}
		
		if("indexPage".equals(pageMode)) {
			
			String ipk=(String) context.get("idxPageKey");
			
			Object ipkv=list.get(list.size()-1).get(ipk.toLowerCase());
			
			if(ipkv instanceof String) {
				ipkv="'"+ipkv+"'";
			}
			
			context.put(PAGED_INDEX, ipkv);
			logInfo(ins,"execute page key:{} value:{}",ipk,ipkv);
			String frag=(String) context.get("idxPageFrag");
			context.put("idxPage", frag);
			
		}
		
		if(CollectionUtils.isNotEmpty(list)) {
			
		if(ins.getTrans()!=null&& CollectionUtils.isNotEmpty(list)) {	
			
		ins.getTrans().transform(list, columnMappingPairs);
		
		//logInfo(ins,"execute Trans for page:{},{}",context.get(START_OFFSET),context.get(END_OFFSET));
		} 
			
		int wnum=writeAgent.execute(list, columnMappingPairs, context);
		
		context.put(CHECKROWDATA, list.get(list.size()-1));
		
		return  wnum;

		}
		return 0;
		
		
		
	}
	
	
	private Integer rwBufferLoad(String sourceSql,JobExecInstanceContext ins ,Map<String, Object> context,DataSource sSource,DataSource tSource,
			String tagTableName,List<Pair<ColumnInfo, ColumnInfo>> columnMappingPairs,WriteAgent writeAgent) throws Exception{
		
		String pageMode= (String) context.get(PAGED_MODE);
		
		String executeSql=wrapSplExp(sourceSql, context);
				
		logDebug( ins,"execute wraped spl:{}" , executeSql);
		
		int countSize=0;
		
		List<Map<String, Object>> list=null;

		StateOfRWBuffer<Map<String, Object>>  buff = QueryHelper.queryStateOfRWBuffer(sSource, executeSql, 2000);
		
		logInfo(ins,"execute page:{},{}",context.get(START_OFFSET),context.get(END_OFFSET));

		while (!buff.isClosed()) 
		{ 
		  log.info("preparing read");
		  buff.waitToRead(); 
		  log.info("can read");
		  
		  list= buff.getReadData();
		  buff.readed();
		  
		  countSize=+doWrite(ins, list, pageMode, context, writeAgent, columnMappingPairs);

		  log.info("readed , next  buff trun---------------------------- countSize:{}" ,countSize);		  
		  }
		
		
		log.info("buff Closed");
		 
		logInfo(ins,"execute writed page:{},{}",context.get(START_OFFSET),context.get(END_OFFSET));
			
		return countSize;
		
		
	}
	


	private Integer load(String sourceSql,JobExecInstanceContext ins ,Map<String, Object> context,DataSource sSource,DataSource tSource,
			String tagTableName,List<Pair<ColumnInfo, ColumnInfo>> columnMappingPairs,WriteAgent writeAgent) throws Exception{
		
		String pageMode= (String) context.get(PAGED_MODE);
		
		String executeSql=wrapSplExp(sourceSql, context);
				
		logDebug( ins,"execute wraped spl:{}" , executeSql);
		
		int countSize=0;
		
		List<Map<String, Object>> list=null;
		
		list= QueryHelper.queryMapList(sSource, executeSql);
		
		//StateOfRWBuffer<Map<String, Object>>  buff = QueryHelper.queryStateOfRWBuffer(sSource, executeSql, 5000);
		
		logInfo(ins,"execute page:{},{}",context.get(START_OFFSET),context.get(END_OFFSET));

		/*
		 * while (!buff.isClosed()) { try { log.info("preparing read");
		 * buff.waitToRead(); } catch (InterruptedException e) { // TODO Auto-generated
		 * catch block e.printStackTrace(); }
		 * 
		 * log.info("can read");
		 * 
		 * list= buff.getReadData();
		 * 
		 * buff.readed();
		 * 
		 * countSize=+doWrite(ins, list, pageMode, context, writeAgent,
		 * 
		 * columnMappingPairs);
		 * 
		 * 
		 * log.info("readed , next  buff trun---------------------------- countSize:{}"
		 * ,countSize);
		log.info("buff Closed");
		 * 
		 * }
		 */
		
		doWrite(ins, list, pageMode, context, writeAgent, columnMappingPairs);
		
		
		
		logInfo(ins,"execute writed page:{},{}",context.get(START_OFFSET),context.get(END_OFFSET));
			
		
		return countSize;

	}
	
	
	private String wrapSplExp(String expSql,Map<String , Object> context) {
		
		EvaluationContext evalContext=expressionHelper.map2EvalContext(context);

		String  wrapedSql=	expressionHelper.parseExpression(expSql).getValue(evalContext, String.class);
		
		return wrapedSql;
		
	}
	


	@Override
	@Autowired
	public void setExecInsPersistService(CheckPointService eipsvc) {
		// TODO Auto-generated method stub
		this.checkPointService=eipsvc;
	}
	
	
	private void logDebug(JobExecInstanceContext ins,String msg,Object...param) {
		msg="Job:{} ,instance:{} ,"+msg;
		log.debug(msg, ins.getJob().getJobName(),ins.getId(), param);
		
	}
	
	private void logInfo(JobExecInstanceContext ins,String msg,Object...param) {
		
		msg="Job:{} ,instance:{} ,"+msg;
		log.info(msg, ins.getJob().getJobName(),ins.getId(), param);
		
	}






	

}
